package com.gjy.kafka.k10;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.header.Header;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

/**
 * 生产者功能：
 * <p>
 * 同步/异步发送
 * 带回调的发送
 * 指定分区发送
 * 带headers发送
 * 批量发送
 * 事务消息
 *
 * @author gjy
 * @version 1.0
 * @since 2025-10-03 11:08:12
 */
public class KafkaProducerService {

    private static final Logger logger = LoggerFactory.getLogger(KafkaProducerService.class);

    private final Producer<String, String> producer;

    public KafkaProducerService(Properties props) {
        this.producer = new KafkaProducer<>(props);
    }

    /**
     * 1. 同步发送消息
     */
    public RecordMetadata sendSync(String topic, String key, String value) {
        try {
            ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, value);
            Future<RecordMetadata> future = producer.send(record);
            return future.get(); // 同步等待结果
        } catch (InterruptedException | ExecutionException e) {
            logger.error("同步发送消息失败: {}", e.getMessage());
            throw new RuntimeException("发送消息失败", e);
        }
    }

    /**
     * 2. 异步发送消息（不带回调）
     */
    public void sendAsync(String topic, String key, String value) {
        ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, value);
        producer.send(record);
    }

    /**
     * 3. 异步发送消息（带回调）
     */
    public void sendAsyncWithCallback(String topic, String key, String value) {
        ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, value);

        producer.send(record, new Callback() {
            @Override
            public void onCompletion(RecordMetadata metadata, Exception exception) {
                if (exception != null) {
                    logger.error("异步发送消息失败: {}", exception.getMessage());
                } else {
                    logger.info("消息发送成功: topic={}, partition={}, offset={}",
                            metadata.topic(), metadata.partition(), metadata.offset());
                }
            }
        });
    }

    /**
     * 4. 发送消息到指定分区
     */
    public RecordMetadata sendToPartition(String topic, Integer partition, String key, String value) {
        try {
            ProducerRecord<String, String> record = new ProducerRecord<>(topic, partition, key, value);
            Future<RecordMetadata> future = producer.send(record);
            return future.get();
        } catch (InterruptedException | ExecutionException e) {
            logger.error("发送消息到分区失败: {}", e.getMessage());
            throw new RuntimeException("发送消息失败", e);
        }
    }

    /**
     * 5. 发送带headers的消息
     */
    public void sendWithHeaders(String topic, String key, String value, List<Header> headers) {
        ProducerRecord<String, String> record = new ProducerRecord<>(topic, null, key, value, headers);

        producer.send(record, new Callback() {
            @Override
            public void onCompletion(RecordMetadata metadata, Exception exception) {
                if (exception != null) {
                    logger.error("发送带headers消息失败: {}", exception.getMessage());
                } else {
                    logger.info("带headers消息发送成功");
                }
            }
        });
    }

    /**
     * 6. 批量发送消息
     */
    public void sendBatch(String topic, List<String> messages) {
        for (int i = 0; i < messages.size(); i++) {
            ProducerRecord<String, String> record = new ProducerRecord<>(topic, null, "key-" + i, messages.get(i));
            producer.send(record);
        }
        producer.flush(); // 确保所有消息都发送
    }

    /**
     * 7. 发送事务消息
     */
    public void sendTransactional(String topic, String key, String value) {
        // 初始化事务
        producer.initTransactions();

        try {
            producer.beginTransaction();

            ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, value);
            producer.send(record);

            // 提交事务
            producer.commitTransaction();
            logger.info("事务消息发送成功");

        } catch (Exception e) {
            producer.abortTransaction();
            logger.error("事务消息发送失败: {}", e.getMessage());
            throw new RuntimeException("事务发送失败", e);
        }
    }

    /**
     * 8. 关闭生产者
     */
    public void close() {
        if (producer != null) {
            producer.close();
            logger.info("Kafka生产者已关闭");
        }
    }

}
